package com.alibaba.cirtus.ots.application.adapter.ability.facade.serviceV2.kafka;

import com.alibaba.cz.base.tool.log.Log;
import com.alibaba.fastjson.JSON;
import com.alibaba.lattice2.epoch.sdk.util.SpringBeanContextUtil;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * yufeng
 * 2023/6/1
 **/
public class KafkaUtils {
    private static KafkaTemplate kafkaTemplate = SpringBeanContextUtil.getBean(KafkaTemplate.class);
    private static final Log log = Log.getLogger(KafkaUtils.class);
    public static KafkaReEntity send(String topic, String str) {
        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(topic, str);
        try {
            log.info("============kafka_send_start_topic:"+topic);
            SendResult<Integer, String> integerStringSendResult = future.get();
            log.info("============kafka_send_start_endresult:"+ JSON.toJSONString(integerStringSendResult));
            int partition = integerStringSendResult.getRecordMetadata().partition();
            long offset = integerStringSendResult.getRecordMetadata().offset();
            KafkaReEntity kafkaReEntity = new KafkaReEntity(partition, offset);
            return kafkaReEntity;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new KafkaReEntity();
    }
    public static KafkaReEntity send(String topic, String key,String str) {
        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(topic,key, str);
        try {
            log.info("============kafka_send_start_topic:"+topic);
            SendResult<Integer, String> integerStringSendResult = future.get();
            log.info("============kafka_send_start_endresult:"+ JSON.toJSONString(integerStringSendResult));
            int partition = integerStringSendResult.getRecordMetadata().partition();
            long offset = integerStringSendResult.getRecordMetadata().offset();
            KafkaReEntity kafkaReEntity = new KafkaReEntity(partition, offset);
            return kafkaReEntity;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new KafkaReEntity();
    }
}

